package cn.lsh.hbase.coprocessor;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.coprocessor.ColumnInterpreter;
import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * 自定义AggregationClient的列解析器
 */
public class MyStringIntColumnInterpreter extends ColumnInterpreter<Long, Long, HBaseProtos.EmptyMsg, HBaseProtos.LongMsg, HBaseProtos.LongMsg> {
	@Override
	public Long getValue(byte[] bytes, byte[] bytes1, Cell cell) throws IOException {
		return Long.valueOf(Bytes.toString(CellUtil.cloneValue(cell)));
	}

	@Override
	public Long add(Long aLong, Long s1) {
		//加法，进行sum运算时会调用
		if (aLong == null) {
			return s1;
		}
		if (s1 == null) {
			return aLong;
		}
		return aLong + s1;
	}

	@Override
	public Long getMaxValue() {
		return Long.MAX_VALUE;
	}

	@Override
	public Long getMinValue() {
		return Long.MIN_VALUE;
	}

	@Override
	public Long multiply(Long aLong, Long s1) {
		//乘法
		if (aLong == null) {
			return s1;
		}
		if (s1 == null) {
			return aLong;
		}
		return aLong * s1;
	}

	@Override
	public Long increment(Long aLong) {
		//增量值
		return aLong == null ? null : aLong + 1;
	}

	@Override
	public Long castToReturnType(Long aLong) {
		return aLong;
	}

	@Override
	public int compare(Long aLong, Long t1) {
		//比较
		return aLong == null ? -1 : t1 == null ? 1 : aLong.compareTo(t1);
	}

	@Override
	public double divideForAvg(Long aLong, Long aLong2) {
		//除法
		return (aLong == null || aLong2 == null) ? Double.NaN : (aLong.doubleValue() / aLong2
				.doubleValue());
	}

	@Override
	public HBaseProtos.EmptyMsg getRequestData() {
		return HBaseProtos.EmptyMsg.getDefaultInstance();
	}

	@Override
	public void initialize(HBaseProtos.EmptyMsg emptyMsg) {

	}

	@Override
	public HBaseProtos.LongMsg getProtoForCellType(Long aLong) {
		return HBaseProtos.LongMsg.newBuilder().setLongMsg(aLong).build();
	}

	@Override
	public Long getCellValueFromProto(HBaseProtos.LongMsg longMsg) {
		return longMsg.getLongMsg();
	}

	@Override
	public HBaseProtos.LongMsg getProtoForPromotedType(Long aLong) {
		return HBaseProtos.LongMsg.newBuilder().setLongMsg(aLong).build();
	}

	@Override
	public Long getPromotedValueFromProto(HBaseProtos.LongMsg longMsg) {
		return longMsg.getLongMsg();
	}

	@Override
	public Long castToCellType(Long aLong) {
		return aLong;
	}

}
